#include "StdAfx.h"

#include <PSDFCore/Data/DataCenter.h>


//---------------------------------------
// MemoryReleaser
//---------------------------------------
void DataCenter::MemoryReleaser::run()
{
	while (!_shouldQuit)
	{
		clearMainBuffer();
		clearAssistBuffer();
		PosixThread::msleep(10000); // sleep for 10 seconds
	}
}

void DataCenter::MemoryReleaser::clearMainBuffer()
{
	for (int i = MAIN_DATABUFFER_LENGTH-1; i >= 0; --i)
	{
		_center->_mainBufferLock[i].lock();

		if (_center->_mainDataBuffer[i].buffer == NULL)
		{
			_center->_mainBufferLock[i].unlock();
			continue;
		}
	
		if (_center->_mainDataBuffer[i].refCount > 0)
		{
			_center->_mainBufferLock[i].unlock();
			break;
		}

		clock_t now = clock();
		if (1.0 * (now - _center->_mainDataBuffer[i].timestamp) / CLOCKS_PER_SEC < 1)
		{
			_center->_mainBufferLock[i].unlock();
			break;
		}

		// this space has not been used for over 1 second, so release the space
		delete [] _center->_mainDataBuffer[i].buffer;
		_center->_mainDataBuffer[i].buffer = NULL;
		_center->_mainDataBuffer[i].bufferLength = 0;
		_center->_mainDataBuffer[i].dataLength = 0;
		_center->_mainDataBuffer[i].timestamp = 0;

		_center->_mainBufferLock[i].unlock();
	}
		
}

void DataCenter::MemoryReleaser::clearAssistBuffer()
{
	_center->_assistBufferLock.lock();

	int i;
	for (i = (int)_center->_assistDataBuffer.size() - 1; i > 0; --i)
	{
		if (_center->_assistDataBuffer[i].refCount > 0) { break; }

		delete [] _center->_assistDataBuffer[i].buffer;
		_center->_assistDataBuffer[i].buffer = NULL;
		_center->_assistDataBuffer[i].bufferLength = 0;
		_center->_assistDataBuffer[i].dataLength = 0;
		_center->_assistDataBuffer[i].timestamp = 0;
	}

	if (i == -1)
	{
		_center->_assistDataBuffer.clear();
	}

	_center->_assistBufferLock.unlock();
}



//----------------------------------
// DataCenter
//----------------------------------
DataCenter::DataCenter()
{
	_memoryReleaser._center = this;
	_memoryReleaser.start();
}

DataCenter::~DataCenter()
{
	_memoryReleaser._shouldQuit = true;

	for (int i = 0; i < MAIN_DATABUFFER_LENGTH; ++i)
	{
		_mainBufferLock[i].lock();
		if (_mainDataBuffer[i].buffer)
		{
			delete [] _mainDataBuffer[i].buffer;
			_mainDataBuffer[i].buffer = NULL;
			_mainDataBuffer[i].bufferLength = 0;
		}
		_mainBufferLock[i].unlock();
	}

	_assistBufferLock.lock();

	for (unsigned int i = 0; i < _assistDataBuffer.size(); ++i)
	{
		if (_assistDataBuffer[i].buffer)
		{
			delete [] _assistDataBuffer[i].buffer;
			_assistDataBuffer[i].buffer = NULL;
			_assistDataBuffer[i].bufferLength = 0;
		}
	}
	_assistDataBuffer.clear();
	
	_assistBufferLock.unlock();

	PosixThread::waitUntilDone(_memoryReleaser.getThreadId());
}

DataCenter* DataCenter::inst()
{
	static DataCenter instance;
	return &instance;
}

int DataCenter::addData( char* data, int length )
{
	int index = addDataToMainBuffer(data, length);
	if (index == -1)
	{
		index = addDataToAssistBuffer(data, length);
	}
	return index;
}

bool DataCenter::clearData( ReferencedData& refData, int newDataLength )
{
	if (refData.refCount > 0)
	{
		return false;
	}

	clock_t now = clock();
	if (1.0 * (now - refData.timestamp) / CLOCKS_PER_SEC < 0.03) // keep all data for 30ms
	{
		return false;
	}

	if (newDataLength > refData.bufferLength)
	{
		delete [] refData.buffer;	
		refData.buffer = new char[newDataLength];
		refData.bufferLength = newDataLength;
	}
	memset(refData.buffer, 0, refData.bufferLength);

	refData.refCount = 0;
	refData.dataLength = 0;
	refData.timestamp = 0;

	return true;
}

int DataCenter::addDataToMainBuffer(char* data, int length)
{
	int i = 0;
	bool ok = false;
	for (; i < MAIN_DATABUFFER_LENGTH; ++i)
	{
		_mainBufferLock[i].lock();
		if (clearData(_mainDataBuffer[i], length))
		{
			memcpy(_mainDataBuffer[i].buffer, data, length);
			_mainDataBuffer[i].dataLength = length;
			_mainDataBuffer[i].timestamp = clock();

			ok = true;
			_mainBufferLock[i].unlock();
			break;
		}
		_mainBufferLock[i].unlock();
	}

	if (!ok) { return -1; }
	else { return i; }
}

int DataCenter::addDataToAssistBuffer(char* data, int length)
{
	unsigned int i = 0;
	bool ok = false;

	for (; i < _assistDataBuffer.size(); ++i)
	{
		_assistBufferLock.lock();
		if (clearData(_assistDataBuffer[i], length))
		{
			memcpy(_assistDataBuffer[i].buffer, data, length);
			_assistDataBuffer[i].dataLength = length;
			_assistDataBuffer[i].timestamp = clock();
			ok = true;
			_assistBufferLock.unlock();
			break;
		}
		_assistBufferLock.unlock();
	}

	if (!ok)
	{
		ReferencedData newData;
		newData.buffer = new char[length];
		memcpy(newData.buffer, data, length);
		newData.bufferLength = length;
		newData.dataLength = length;
		newData.timestamp = clock();
		_assistDataBuffer.push_back(newData);
	}

	return i;
}